package rxgo

import (
	"context"
	"fmt"
	"reflect"
	"time"
)

// filtering node implementation of streamOperator
type filteringOperator struct {
	opFunc func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool)
}

//op op函数
func (sop filteringOperator) op(ctx context.Context, o *Observable) {
	// must hold defintion of flow resourcs here, such as chan etc., that is allocated when connected
	// this resurces may be changed when operation routine is running.
	in := o.pred.outflow
	out := o.outflow
	//fmt.Println(o.name, "operator in/out chan ", in, out)

	// Scheduler
	go func() {
		end := false

		for x := range in {
			if end {
				break
			}

			// can not pass a interface as parameter (pointer) to gorountion for it may change its value outside!
			temp := reflect.ValueOf(x)
			// send an error to stream if the flip not accept error
			err, ok := x.(error)
			if ok && !o.flip_accept_error {
				o.sendToFlow(ctx, err, out)
				continue
			}

			if sop.opFunc(ctx, o, temp, out) {
				end = true
			}
		}

		if o.flip != nil {
			buffer := (reflect.ValueOf(o.flip))
			for i := 0; i < buffer.Len(); i++ {
				o.sendToFlow(ctx, buffer.Index(i).Interface(), out)
			}
		}

		o.closeFlow(out)
	}()
}

//newFilterObservable 新建一个Filter Observabl
func (parent *Observable) newFilterObservable(name string) (o *Observable) {
	//new Observable
	o = newObservable()
	o.Name = name

	//chain Observables
	parent.next = o
	o.pred = parent
	o.root = parent.root

	//set options
	o.buf_len = BufferLen
	return o
}

var count = 0
var tempTime time.Duration

//Debounce 按时间防抖动
func (parent *Observable) Debounce(timespan time.Duration) (o *Observable) {
	tempTime = timespan
	o = parent.newFilterObservable("debounce")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	count = 0
	o.operator = debounceOp
	return o
}

var debounceOp = filteringOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		count++
		go func() {
			tempCount := count
			time.Sleep(tempTime)
			select {
			case <-ctx.Done():
				return
			default:
				if tempCount == count {
					o.sendToFlow(ctx, item.Interface(), out)
				}
			}
		}()
		return false
	},
}

var tempMap map[string]bool

//Distinct 过滤掉重复出现的元素
func (parent *Observable) Distinct() (o *Observable) {
	o = parent.newFilterObservable("distinct")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	tempMap = map[string]bool{}
	o.operator = distinctOp
	return o
}

var distinctOp = filteringOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		itemStr := fmt.Sprintf("%v", item)
		_, ok := tempMap[itemStr]
		if !ok {
			tempMap[itemStr] = true
			o.sendToFlow(ctx, item.Interface(), out)
		}
		return false
	},
}

var tempNum int

//ElementAt 取第几个元素
func (parent *Observable) ElementAt(num int) (o *Observable) {
	tempNum = num
	o = parent.newFilterObservable("elementAt")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	count = 0
	o.operator = elementAtOp

	return o
}

var elementAtOp = filteringOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		if count == tempNum {
			o.sendToFlow(ctx, item.Interface(), out)
			return true
		}
		count++
		return false
	},
}

// Filter `func(x anytype) bool` filters items in the original Observable and returns
// a new Observable with the filtered items.
func (parent *Observable) Filter(f interface{}) (o *Observable) {
	// check validation of f
	fv := reflect.ValueOf(f)
	inType := []reflect.Type{typeAny}
	outType := []reflect.Type{typeBool}
	b, ctx_sup := checkFuncUpcast(fv, inType, outType, true)
	if !b {
		panic(ErrFuncFlip)
	}

	o = parent.newTransformObservable("filter")
	o.flip_accept_error = checkFuncAcceptError(fv)

	o.flip_sup_ctx = ctx_sup
	o.flip = fv.Interface()
	o.operator = filterOperater
	return o
}

var filterOperater = transOperater{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

	fv := reflect.ValueOf(o.flip)
	var params = []reflect.Value{x}
	rs, skip, stop, e := userFuncCall(fv, params)

	var item interface{} = rs[0].Interface()
	if stop {
		end = true
		return
	}
	if skip {
		return
	}
	if e != nil {
		item = e
	}
	// send data
	if !end {
		if b, ok := item.(bool); ok && b {
			end = o.sendToFlow(ctx, x.Interface(), out)
		}
	}

	return
}}

//First 完成时返回第一个元素
func (parent *Observable) First() (o *Observable) {
	o = parent.newFilterObservable("first")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.operator = firstOp
	return o
}

var firstOp = filteringOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		o.sendToFlow(ctx, item.Interface(), out)
		return true
	},
}

//IgnoreElements 忽略所有元素
func (parent *Observable) IgnoreElements() (o *Observable) {
	o = parent.newFilterObservable("ignoreElements")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.operator = ignoreElementsOp
	return o
}

var ignoreElementsOp = filteringOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		return false
	},
}

//Last 完成时返回最后一个元素
func (parent *Observable) Last() (o *Observable) {
	o = parent.newFilterObservable("last")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.operator = lastOp
	return o
}

var lastOp = filteringOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		o.flip = append([]interface{}{}, item.Interface())
		return false
	},
}

var tempSample chan interface{}

//Sample 定期发射Observable最近发射的数据项
func (parent *Observable) Sample(sample chan interface{}) (o *Observable) {
	tempSample = sample
	o = parent.newFilterObservable("sample")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.operator = sampleOP
	return o
}

var sampleOP = filteringOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		var latest interface{} = nil
		latest = item.Interface()
		go func() {
			tempEnd := true
			for tempEnd {
				select {
				case <-ctx.Done():
					tempEnd = true
				case <-tempSample:
					if latest != nil {
						if o.sendToFlow(ctx, latest, out) {
							tempEnd = false
						}
						latest = nil
					}
				}
			}
		}()
		return false
	},
}

//Skip 跳过前n个数据
func (parent *Observable) Skip(num int) (o *Observable) {
	tempNum = num
	o = parent.newFilterObservable("skip")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	count = 0
	o.operator = skipOp

	return o
}

var skipOp = filteringOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		count++
		if count > tempNum {
			o.sendToFlow(ctx, item.Interface(), out)
		}
		return false
	},
}

var tempLasts []interface{}

//SkipLast 跳过最后n个数据
func (parent *Observable) SkipLast(num int) (o *Observable) {
	tempNum = num
	o = parent.newFilterObservable("skipLast")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	count = 0
	o.operator = skipLastOp

	return o
}

var skipLastOp = filteringOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		//var lasts []interface{}
		if count == tempNum {
			o.sendToFlow(ctx, tempLasts[0], out)
			tempLasts = tempLasts[1:]
		} else {
			count++
		}
		tempLasts = append(tempLasts, item.Interface())
		return false
	},
}

//Take 取前n个数据
func (parent *Observable) Take(num int) (o *Observable) {
	tempNum = num
	o = parent.newFilterObservable("take")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	count = 0
	o.operator = takeOp

	return o
}

var takeOp = filteringOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		count++
		if count > tempNum {
			return true
		}
		o.sendToFlow(ctx, item.Interface(), out)
		return false
	},
}

var tempLasts2 []interface{}

//TakeLast 取最后n个数据
func (parent *Observable) TakeLast(num int) (o *Observable) {
	tempNum = num
	o = parent.newFilterObservable("takeLast")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	count = 0

	o.operator = takeLastOp

	return o
}

var takeLastOp = filteringOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		count++
		if count <= tempNum {
			tempLasts2 = append(tempLasts2, item.Interface())
		} else {
			tempLasts2 = tempLasts2[1:]
			tempLasts2 = append(tempLasts2, item.Interface())
		}
		o.flip = tempLasts2
		return false
	},
}
